package com.sardine.base.hbase.executor;

import com.sardine.base.hbase.exception.HException;
import com.sardine.base.hbase.persister.HEntity;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 上海京东到家元信信息技术有限公司<br>
 * Created with IntelliJ IDEA.
 * Date:2017/11/22
 * Time:16:38
 *
 * @author daifeng<br>
 *         daifeng@imdada.cn
 * @since JDK 1.7 <br>
 * Copyright (c) 2017 imdada System Incorporated All Rights Reserved.
 */
public class DefaultHBaseRepository implements HBaseRepository {
    private static final Logger logger = LoggerFactory.getLogger(DefaultHBaseRepository.class);
    private Connection connection;

    public DefaultHBaseRepository(Connection connection) {
        this.connection = connection;
    }

    /**
     * 保存一个实体
     *
     * @param hEntity
     * @return
     */
    @Override
    public void save(final HEntity hEntity) {
        final String tableSpace = hEntity.getTableSpace();
        final String tableName = hEntity.getTableName();
        this.execute(new HCallBack<Void>() {
            @Override
            public Void doHBase(Table table) throws IOException {
                String rowKey = hEntity.getRowKey();
                List<HEntity.DataModel> dataModelList = hEntity.getDataModelList();
                Put put = new Put(Bytes.toBytes(rowKey));
                for (HEntity.DataModel dataModel : dataModelList) {
                    put.addColumn(Bytes.toBytes(dataModel.getColumnFamily()), Bytes.toBytes(dataModel.getQualifier()), Bytes.toBytes(dataModel.getFieldValue()));
                }
                table.put(put);
                return null;
            }
        }, tableSpace, tableName);
    }

    /**
     * 批量插入
     *
     * @param hEntities
     */
    @Override
    public void save(final List<HEntity> hEntities) {
        HEntity hEntity = hEntities.get(0);
        final String tableSpace = hEntity.getTableSpace();
        final String tableName = hEntity.getTableName();
        this.execute(new HCallBack<Void>() {
            @Override
            public Void doHBase(Table table) throws IOException {
                List<Put> puts = new ArrayList<Put>();
                for (HEntity entity : hEntities) {
                    String rowKey = entity.getRowKey();
                    List<HEntity.DataModel> dataModelList = entity.getDataModelList();
                    Put put = new Put(Bytes.toBytes(rowKey));
                    for (HEntity.DataModel dataModel : dataModelList) {
                        put.addColumn(Bytes.toBytes(dataModel.getColumnFamily()), Bytes.toBytes(dataModel.getQualifier()), Bytes.toBytes(dataModel.getFieldValue()));
                    }
                    puts.add(put);
                }
                table.put(puts);
                return null;
            }
        }, tableSpace, tableName);
    }

    /**
     * 异步
     *
     * @param hEntities
     */
    @Override
    public void saveAsync(List<HEntity> hEntities) {
        BufferedMutator bufferedMutator = null;
        try {
            HEntity hEntity = hEntities.get(0);
            final String tableSpace = hEntity.getTableSpace();
            final String tableName = hEntity.getTableName();
            bufferedMutator = connection.getBufferedMutator(TableName.valueOf(tableSpace + ":" + tableName));
            List<Put> puts = new ArrayList<Put>();
            for (HEntity entity : hEntities) {
                String rowKey = entity.getRowKey();
                List<HEntity.DataModel> dataModelList = entity.getDataModelList();
                Put put = new Put(Bytes.toBytes(rowKey));
                for (HEntity.DataModel dataModel : dataModelList) {
                    put.addColumn(Bytes.toBytes(dataModel.getColumnFamily()), Bytes.toBytes(dataModel.getQualifier()), Bytes.toBytes(dataModel.getFieldValue()));
                }
                puts.add(put);
            }
            bufferedMutator.mutate(puts);//批量异步写
        } catch (IOException e) {
            throw new HException(e);
        } finally {
            if (bufferedMutator != null) {
                try {
                    bufferedMutator.close();
                } catch (IOException e) {
                    throw new HException("close  hbase table error", e);
                }
            }
        }
    }

    /**
     * 根据rowKey 查询一个值
     *
     * @param hEntity
     * @return
     */
    @Override
    public HEntity load(final HEntity hEntity) {
        final String tableSpace = hEntity.getTableSpace();
        final String tableName = hEntity.getTableName();
        HEntity execute = this.execute(new HCallBack<HEntity>() {
            @Override
            public HEntity doHBase(Table table) throws IOException {
                String rowKey = hEntity.getRowKey();
                Get get = new Get(Bytes.toBytes(rowKey));
                Result result = table.get(get);
                if (result != null) {
                    List<HEntity.DataModel> dataModelList = hEntity.getDataModelList();
                    List<Cell> cells = result.listCells();
                    if (cells != null) {
                        for (Cell cell : cells) {
                            HEntity.DataModel model = new HEntity.DataModel();
                            model.setColumnFamily(Bytes.toString(CellUtil.cloneFamily(cell)));
                            model.setQualifier(Bytes.toString(CellUtil.cloneQualifier(cell)));
                            model.setFieldValue(Bytes.toString(CellUtil.cloneValue(cell)));
                            dataModelList.add(model);
                        }
                    }
                }
                return hEntity;
            }
        }, tableSpace, tableName);

        return execute;
    }

    /**
     * 批量查询一组数据
     *
     * @param hEntity
     * @param rowKey
     * @return
     */
    @Override
    public List<HEntity> loadList(final HEntity hEntity, final List<String> rowKey) {
        final String tableSpace = hEntity.getTableSpace();
        final String tableName = hEntity.getTableName();
        List<HEntity> execute = this.execute(new HCallBack<List<HEntity>>() {
            @Override
            public List<HEntity> doHBase(Table table) throws IOException {
                List<Get> getList = new ArrayList<Get>();
                for (String key : rowKey) {
                    Get get5 = new Get(Bytes.toBytes(key));
                    getList.add(get5);
                }
                Result[] results = table.get(getList);
                List<HEntity> entities = new ArrayList<>();
                for (Result result : results) {
                    if (result != null) {
                        HEntity ent = new HEntity();
                        ent.setTableName(tableName);
                        ent.setTableSpace(tableSpace);
                        List<HEntity.DataModel> dataModelList = ent.getDataModelList();
                        List<Cell> cells = result.listCells();
                        if (cells != null) {
                            for (Cell cell : cells) {
                                HEntity.DataModel model = new HEntity.DataModel();
                                model.setColumnFamily(Bytes.toString(CellUtil.cloneFamily(cell)));
                                model.setQualifier(Bytes.toString(CellUtil.cloneQualifier(cell)));
                                model.setFieldValue(Bytes.toString(CellUtil.cloneValue(cell)));
                                dataModelList.add(model);
                            }
                        }
                        entities.add(ent);
                    }
                }
                return entities;
            }
        }, tableSpace, tableName);
        return execute;
    }

    /**
     * 执行
     *
     * @param callBack
     * @param tableSpace
     * @param tableName
     * @param <T>
     * @return
     */
    private <T> T execute(HCallBack<T> callBack, String tableSpace, String tableName) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableSpace + ":" + tableName));
            return callBack.doHBase(table);
        } catch (IOException e) {
            throw new HException(e);
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    throw new HException("close  hbase table error", e);
                }
            }
        }
    }

    abstract class HCallBack<T> {
        public abstract T doHBase(Table table) throws IOException;
    }
}
